﻿using CanalService.Model;
using CanalSharp.Connections;
using CanalSharp.Protocol;
using Google.Protobuf.Collections;
using MongoDB.Driver;
using System.Reflection;

namespace CanalService.Services;

/// <summary>
/// Canal数据同步MongoDb
/// </summary>
public class CanalToMongoDbService
{

    private readonly string _mongoConn;

    public CanalToMongoDbService()
    {
        _mongoConn = "http://localhost:27017";
    }

    /// <summary>
    /// 监听服务
    /// </summary>
    public async static void OnStart()
    {
        //1、日志准备
        var loggerFactory = LoggerFactory.Create(builder =>
        {
            builder
               .AddFilter("Microsoft", LogLevel.Debug)
               .AddFilter("System", LogLevel.Information)
               .AddConsole();
        });


        //1.1 监听Canal的调用日志
        var logger = loggerFactory.CreateLogger<SimpleCanalConnection>();

        //2、建立Canal服务器连接，默认端口是11111，
        var canalConn = new SimpleCanalConnection(new SimpleCanalOptions("127.0.0.1", 11111, "101"), logger);
        await canalConn.ConnectAsync();

        //3、订阅数据库和表
        await canalConn.SubscribeAsync("ndc_test.canal_table");

        //4、读取Canal数据
        while (true)
        {
            try
            {
                //4.1 分段读取Canal数据
                var canalData = await canalConn.GetWithoutAckAsync(1024);

                //4.2 判断数据类型(加锁或不加锁)
                var entries = canalData.Entries;
                foreach (var entry in entries)
                {
                    //不处理事务标记
                    if (entry.EntryType == EntryType.Transactionbegin || entry.EntryType == EntryType.Transactionend)
                    {
                        continue;
                    }
                    //4.3 判断数据操作类型：增删改查
                    RowChange rowChange = RowChange.Parser.ParseFrom(entry.StoreValue);
                    switch (rowChange.EventType)
                    {
                        case EventType.Compatibleproto2:
                            break;
                        case EventType.Insert:
                            logger.LogInformation("============增加数据============");
                            await HandleCanalInsertDataAsync(rowChange.RowDatas, entry);
                            break;
                        case EventType.Update:
                            logger.LogInformation("===========修改数据=============");
                            break;
                        case EventType.Delete:
                            logger.LogInformation("===========删除数据=============");
                            break;
                        case EventType.Create:
                            break;
                        case EventType.Alter:
                            break;
                        case EventType.Erase:
                            break;
                        case EventType.Query:
                            logger.LogInformation("===========查询数据=============");
                            //await HandleCanalInsertDataAsync(rowChange.RowDatas, entry);
                            break;
                        case EventType.Truncate:
                            break;
                        case EventType.Rename:
                            break;
                        case EventType.Cindex:
                            break;
                        case EventType.Dindex:
                            break;
                        case EventType.Gtid:
                            break;
                        case EventType.Xacommit:
                            break;
                        case EventType.Xarollback:
                            break;
                        case EventType.Mheartbeat:
                            break;
                        default:
                            break;
                    }

                }

            }
            catch (Exception)
            {
                throw;
            }  
        }
    }

    /// <summary>
    /// 处理Canal新增数据变动
    /// </summary>
    /// <returns></returns>
    private static async Task HandleCanalInsertDataAsync(RepeatedField<RowData> rowDatas, Entry entry)
    {
        foreach (RowData rowData in rowDatas)
        {
            //1.获取需要同步的业务数据
            List<Column> columns = rowData.AfterColumns.ToList();
            //2.获取数据库
            var schemaName = entry.Header.SchemaName;
            //3.获取表
            var tableName = entry.Header.TableName;
            //4.根据表名走不同业务
            if(tableName == "usertest")
            {
                await Console.Out.WriteLineAsync("同步usertest表。。。");
            }
            if(tableName == "canal_table")
            {
                await Console.Out.WriteLineAsync("同步canal_table表");

                InsertUserTestToMongoDb(columns);
            }
        }
    }


    /// <summary>
    /// 同步数据到Mongo
    /// </summary>
    /// <param name="columns"></param>
    private static void InsertUserTestToMongoDb(List<Column> columns)
    {
        try
        { 
            CanalTable model = new();
            // 1、数据转换
            // Type
            System.Type type = typeof(CanalTable);
            // 2、取出具体数据
            foreach (var column in columns)
            {
                // 3、取出字段名
                string FiledName = column.Name;
                // 4、取出字段值
                string FiledValue = column.Value;
                Console.WriteLine(FiledName + ":" + FiledValue);
                // 5、属性赋值
                PropertyInfo propertyInfo = type.GetProperty(FiledName);
                // 6、数据类型转换
                if (propertyInfo != null)
                {
                    System.Type type1 = propertyInfo.PropertyType;
                    if (type1 == typeof(int))
                    {
                        propertyInfo.SetValue(model, int.Parse(FiledValue));
                    }
                    else if (type1 == typeof(string))
                    {
                        propertyInfo.SetValue(model, FiledValue);
                    }
                    else if (type1 == typeof(double))
                    {
                        propertyInfo.SetValue(model, double.Parse(FiledValue));
                    }
                    else if (type1 == typeof(bool))
                    {
                        propertyInfo.SetValue(model, bool.Parse(FiledValue));
                    }
                    else if (type1 == typeof(DateTime))
                    {
                        propertyInfo.SetValue(model, DateTime.Parse(FiledValue));
                    }
                    else
                    {
                        Console.WriteLine("类型错误");
                    }
                }
            }

            // canal数据同步
            // 1、建立MongoDB连接
            var client = new MongoClient("mongodb://localhost:27017");

            // 2、创建数据库
            var database = client.GetDatabase("CanalDataSyncDb");

            // 3、获取用户表(自己创建商品集合)
            IMongoCollection<CanalTable> canalTable = database.GetCollection<CanalTable>("CanalTable");

            // 4、写入MongoDb，完成同步
            canalTable.InsertOne(model);
        }
        catch (Exception e)
        {
            Console.WriteLine(e);
        } 
    }
     /*
      其他操作，如更新、删除基本都类似。。。
      */
}
